package com.by.rpc.transaction;

import com.alibaba.fastjson2.JSON;
import com.by.log.Logger;
import com.by.log.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 基于Redis原子操作+消息队列的分布式事务管理器
 * 适用于秒杀等高并发场景
 */
@Component
public class RedisAtomicTransactionManager {
    // 添加自定义日志实例
    private static final Logger log = LoggerFactory.getLogger(RedisAtomicTransactionManager.class);

    private final RedisTemplate<String, Object> redisTemplate;
    private final RedisDistributedLock distributedLock;

    // 事务准备队列
    private static final String TRANSACTION_PREPARE_QUEUE = "atomic_transaction:prepare";

    // 事务提交队列
    private static final String TRANSACTION_COMMIT_QUEUE = "atomic_transaction:commit";

    // 事务回滚队列
    private static final String TRANSACTION_ROLLBACK_QUEUE = "atomic_transaction:rollback";

    // 事务状态前缀
    private static final String TRANSACTION_STATUS_PREFIX = "atomic_transaction:status:";

    // 事务内容前缀
    private static final String TRANSACTION_CONTENT_PREFIX = "atomic_transaction:content:";

    public RedisAtomicTransactionManager(RedisTemplate<String, Object> redisTemplate,
                                         RedisDistributedLock distributedLock) {
        this.redisTemplate = redisTemplate;
        this.distributedLock = distributedLock;
    }
    
    /**
     * 开始分布式事务
     * @param transactionContent 事务内容
     * @param timeout 超时时间(秒)
     * @return 事务ID
     */
    public String beginTransaction(TransactionContent transactionContent, long timeout) {
        String transactionId = generateTransactionId();
        
        try {
            // 序列化事务内容并存储到Redis
            String contentKey = TRANSACTION_CONTENT_PREFIX + transactionId;
            redisTemplate.opsForValue().set(contentKey, JSON.toJSONString(transactionContent), timeout, TimeUnit.SECONDS);
            
            // 设置事务状态为进行中
            String statusKey = TRANSACTION_STATUS_PREFIX + transactionId;
            redisTemplate.opsForValue().set(statusKey, "TRYING", timeout, TimeUnit.SECONDS);
            
            // 将事务放入准备队列
            TransactionMessage prepareMessage = new TransactionMessage();
            prepareMessage.setTransactionId(transactionId);
            prepareMessage.setTimestamp(System.currentTimeMillis());
            prepareMessage.setType(TransactionMessageType.PREPARE);
            prepareMessage.setContent(transactionContent);
            
            redisTemplate.opsForList().leftPush(TRANSACTION_PREPARE_QUEUE, JSON.toJSONString(prepareMessage));
            
            log.info("Atomic transaction started: {}", transactionId);
            return transactionId;
        } catch (Exception e) {
            log.error("Failed to begin atomic transaction: {}", transactionId, e);
            return null;
        }
    }
    
    /**
     * 提交事务
     * @param transactionId 事务ID
     */
    public void commitTransaction(String transactionId) {
        try {
            // 更新事务状态
            String statusKey = TRANSACTION_STATUS_PREFIX + transactionId;
            redisTemplate.opsForValue().set(statusKey, "COMMITTED", 60, TimeUnit.SECONDS);
            
            // 获取事务内容
            String contentKey = TRANSACTION_CONTENT_PREFIX + transactionId;
            String contentJson = (String) redisTemplate.opsForValue().get(contentKey);
            TransactionContent transactionContent = null;
            if (contentJson != null) {
                transactionContent = JSON.parseObject(contentJson, TransactionContent.class);
            }
            
            // 发送提交消息
            TransactionMessage commitMessage = new TransactionMessage();
            commitMessage.setTransactionId(transactionId);
            commitMessage.setTimestamp(System.currentTimeMillis());
            commitMessage.setType(TransactionMessageType.COMMIT);
            commitMessage.setContent(transactionContent);
            
            redisTemplate.opsForList().leftPush(TRANSACTION_COMMIT_QUEUE, JSON.toJSONString(commitMessage));
            
            // 清理事务内容
            redisTemplate.delete(contentKey);
            
            log.info("Atomic transaction committed: {}", transactionId);
        } catch (Exception e) {
            log.error("Failed to commit atomic transaction: {}", transactionId, e);
        }
    }
    
    /**
     * 回滚事务
     * @param transactionId 事务ID
     */
    public void rollbackTransaction(String transactionId) {
        try {
            // 更新事务状态
            String statusKey = TRANSACTION_STATUS_PREFIX + transactionId;
            redisTemplate.opsForValue().set(statusKey, "ROLLED_BACK", 60, TimeUnit.SECONDS);
            
            // 获取事务内容
            String contentKey = TRANSACTION_CONTENT_PREFIX + transactionId;
            String contentJson = (String) redisTemplate.opsForValue().get(contentKey);
            TransactionContent transactionContent = null;
            if (contentJson != null) {
                transactionContent = JSON.parseObject(contentJson, TransactionContent.class);
            }
            
            // 发送回滚消息
            TransactionMessage rollbackMessage = new TransactionMessage();
            rollbackMessage.setTransactionId(transactionId);
            rollbackMessage.setTimestamp(System.currentTimeMillis());
            rollbackMessage.setType(TransactionMessageType.ROLLBACK);
            rollbackMessage.setContent(transactionContent);
            
            redisTemplate.opsForList().leftPush(TRANSACTION_ROLLBACK_QUEUE, JSON.toJSONString(rollbackMessage));
            
            // 清理事务内容
            redisTemplate.delete(contentKey);
            
            log.info("Atomic transaction rolled back: {}", transactionId);
        } catch (Exception e) {
            log.error("Failed to rollback atomic transaction: {}", transactionId, e);
        }
    }
    
    /**
     * 生成事务ID
     * @return 事务ID
     */
    public String generateTransactionId() {
        return UUID.randomUUID().toString().replace("-", "");
    }
    
    /**
     * 获取分布式锁
     * @param lockKey 锁的key
     * @param expireTime 过期时间(秒)
     * @return 锁标识
     */
    public String acquireLock(String lockKey, long expireTime) {
        return distributedLock.tryLock(lockKey, expireTime);
    }
    
    /**
     * 释放分布式锁
     * @param lockKey 锁的key
     * @param lockValue 锁标识
     * @return 是否释放成功
     */
    public boolean releaseLock(String lockKey, String lockValue) {
        return distributedLock.releaseLock(lockKey, lockValue);
    }
    
    /**
     * 原子性扣减库存
     * @param stockKey 库存key
     * @param quantity 扣减数量
     * @return 扣减后剩余库存，-1表示库存不足
     */
    public long decrementStock(String stockKey, int quantity) {
        return distributedLock.decrementStock(stockKey, quantity);
    }
    
    /**
     * 原子性增加库存
     * @param stockKey 库存key
     * @param quantity 增加数量
     * @return 增加后的库存
     */
    public long incrementStock(String stockKey, int quantity) {
        return distributedLock.incrementStock(stockKey, quantity);
    }
    
    /**
     * 事务消息
     */
    public static class TransactionMessage {
        private String transactionId;
        private long timestamp;
        private TransactionMessageType type;
        private TransactionContent content;
        
        // Getters and setters
        public String getTransactionId() {
            return transactionId;
        }
        
        public void setTransactionId(String transactionId) {
            this.transactionId = transactionId;
        }
        
        public long getTimestamp() {
            return timestamp;
        }
        
        public void setTimestamp(long timestamp) {
            this.timestamp = timestamp;
        }
        
        public TransactionMessageType getType() {
            return type;
        }
        
        public void setType(TransactionMessageType type) {
            this.type = type;
        }
        
        public TransactionContent getContent() {
            return content;
        }
        
        public void setContent(TransactionContent content) {
            this.content = content;
        }
    }
    
    /**
     * 事务内容
     */
    public static class TransactionContent {
        // 业务标识
        private String businessId;
        
        // 操作类型
        private String operationType;
        
        // 操作数据
        private Object data;
        
        // 补偿操作数据
        private Object compensationData;
        
        // Getters and setters
        public String getBusinessId() {
            return businessId;
        }
        
        public void setBusinessId(String businessId) {
            this.businessId = businessId;
        }
        
        public String getOperationType() {
            return operationType;
        }
        
        public void setOperationType(String operationType) {
            this.operationType = operationType;
        }
        
        public Object getData() {
            return data;
        }
        
        public void setData(Object data) {
            this.data = data;
        }
        
        public Object getCompensationData() {
            return compensationData;
        }
        
        public void setCompensationData(Object compensationData) {
            this.compensationData = compensationData;
        }
    }
    
    /**
     * 事务消息类型
     */
    public enum TransactionMessageType {
        /**
         * 准备阶段
         */
        PREPARE,
        
        /**
         * 提交
         */
        COMMIT,
        
        /**
         * 回滚
         */
        ROLLBACK
    }
}